package com.consumer.定向队列消费者;

import com.fasterxml.jackson.databind.JsonNode;
import com.rabbitmq.client.Channel;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.messaging.handler.annotation.Headers;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.util.Map;

@RabbitListener(
        containerFactory = "rabbitListenerContainerFactory",
        errorHandler = "rabbitConsumerListenerErrorHandler",
        queues = {"firstDirectQueue"},
        //如果设置为true，则RabbitListenerContainerFactory中的ConcurrentConsumers必须为1，即本实例中的单个消费者独占队列
        exclusive=false
)
@Component
public class DirectQueueConsumer {
    @RabbitHandler(isDefault=true)
    public void onConsume(@Payload JsonNode body, Channel channel, @Headers Map<String, Object> headers) throws IOException {
        /**
         * prefetchSize=0表示：消息大小是否限制（听说rabbitMQ没实现这个）
         * prefetchCount=1表示：每个customer会在MQ预取一些消息放入内存的LinkedBlockingQueue中
         * 如果对于消息执行的顺序有特别要求，应该设置为1
         * 服务端设置接收broker每次推送消息的个数，设置ack手动签收，当消费端处理完当次推送的消息后，broker会再次推送一定个数的消息。
         * 如果没处理完，就不让broker推送消息，直至消费端成功签收。
         * global=true/false：表示限制是在channel上还是在consumer上（听说rabbitMQ没实现这个）
         */
        //channel.basicQos(prefetchSize, prefetchCount, global);
        channel.basicQos(8,true);
        //multiple：是否批量. true：将一次性ack所有小于deliveryTag的消息。
        //channel.basicAck((Long)headers.get(AmqpHeaders.DELIVERY_TAG), false);

        /**
         * deliveryTag：该消息的index。
         * multiple：是否批量. true：将一次性拒绝所有小于deliveryTag的消息。
         * 定义为true时会把小于deliveryTag的所有消息都ack，所有小于deliveryTag的消息其实就是你qos中限制的个数。
         * requeue：被拒绝的是否重新入队列。
         */
        //channel.basicNack(deliveryTag, multiple, requeue);
        System.out.println("QueueConsumer得到的消息为" + body);
        /**
         * deliveryTag：该消息的index。
         * requeue：被拒绝的是否重新入队列。
         */
        channel.basicReject((Long)headers.get(AmqpHeaders.DELIVERY_TAG), false);
    }
}
